package com.itcm.wc;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 描述: 流处理 - 无界
 * <p>
 * <p>
 * 服务端：
 * Linux中nc命令是一个功能强大的网络工具，全称是netcat
 * 输入命令 nc -lk 9001
 * 然后输入数据
 * <p>
 * 客户端(当前代码)：
 * 监听服务端9001端口 读取文本流
 *
 * @author cm_fighting
 * @create 2022-05-08 下午10:07
 */
public class StreamWordCount {

    public static void main(String[] args) throws Exception {

        // 1. 创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 读取文本流
        /*
            这里的服务端配置信息 也可以从启动参数上获取
            使用ParameterTool.fromArgs(args) 得到parameterTool对象
            启动参数和值是键值对方式存储，通过get方法即可获取启动参数上配置的值
            如：String host = parameterTool.get("host")
         */
        DataStreamSource<String> lineDataStream = env.socketTextStream("127.0.0.1", 9001);

        // 后面步骤和无界流处理一致
        SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOneTuple = lineDataStream.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
            String[] words = line.split(" ");
            for (String word : words) {
                out.collect(Tuple2.of(word, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.LONG));

        KeyedStream<Tuple2<String, Long>, String> wordAndOneKeyedStream = wordAndOneTuple.keyBy(data -> data.f0);

        SingleOutputStreamOperator<Tuple2<String, Long>> sum = wordAndOneKeyedStream.sum(1);

        sum.print();

        env.execute();

        /*
            测试结果 ：服务端输入一条数据，客户端会接收到一条数据并输出处理结果
         */
    }

}
